package com.atzly.community.datasource.config;

import com.atzly.community.datasource.ZookeeperListenerHolder;
import com.atzly.community.datasource.ZookeeperService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.*;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.nio.charset.StandardCharsets;

/**
 * @author zlysimida
 * @date 2022/11/27 - 12:41
 */
@ConfigurationProperties(prefix = "zookeeper")
@Configuration
@Slf4j
public class zookeeperConfig {

    private String connections = "172.16.2.158:2181";
    private Integer sessionTimeoutMs = 50000;
    private Integer connectionTimeoutMs = 30000;
    /**
     * 两次重试之间的间隔时间
     */
    private int baseSleepTimeMs = 1000;
    /**
     * 重试次数
     */
    private int maxRetries = 3;

    private static String username = "zhanglongyan";

    private static String password = "123456";

    @Bean
    public CuratorFramework curatorFramework() {
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString(connections)
                .sessionTimeoutMs(sessionTimeoutMs)
                .connectionTimeoutMs(connectionTimeoutMs)
                .authorization("digest",(username + ":" + password).getBytes())
                .retryPolicy(new ExponentialBackoffRetry(baseSleepTimeMs, maxRetries))
                .build();
        client.start();
        return client;
    }

    public static void main(String[] args) throws Exception {
        ZookeeperService zookeeperService = new ZookeeperService();
        String path = "/test";
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString("192.168.220.134:2181")
                .sessionTimeoutMs(50000)
                .connectionTimeoutMs(30000)
                .authorization("digest",(username + ":" + password).getBytes(StandardCharsets.UTF_8))
                .retryPolicy(new ExponentialBackoffRetry(1000,3))
                .build();
        client.start();

        byte[] bytes = client.getData().forPath(path);
        System.out.println(new String(bytes));


        // 添加监听回调函数
        TreeCacheListener listener = (curatorFramework, event) -> {
            ChildData childData = event.getData();
            if (childData == null || childData.getData() == null) {
                return;
            }
            String path1 = childData.getPath();
            String data = new String(event.getData().getData(), StandardCharsets.UTF_8);
            if (event.getType() == TreeCacheEvent.Type.NODE_ADDED) {
                System.out.println("监听到新增数据库配置：【" + path1 + " : " + data + "】");
                if (!StringUtils.isEmpty(path1)) {
                    byte[] bytes1 = client.getData().forPath(path1);
                    System.out.println(new String(bytes1));
                }
            } else if (event.getType() == TreeCacheEvent.Type.NODE_UPDATED) {
                System.out.println("监听到数据源配置修改：【" + path1 + " : " + data + "】");
                if (!StringUtils.isEmpty(path1)) {
                    byte[] bytes1 = client.getData().forPath(path1);
                    System.out.println(new String(bytes1));
                }
            }
        };

        // 注册监听器
        Thread thread = new Thread(() -> {
            try {
                TreeCache treeCache = TreeCache.newBuilder(client, path).setCacheData(true).build();
                treeCache.getListenable().addListener(listener);
                treeCache.start();
                ZookeeperListenerHolder.ZK_LISTENER_CACHES.put("listenerChange",treeCache);
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
        thread.setDaemon(false);
        thread.start();


        Thread.sleep(1000 * 60 * 10);
    }


}
